package com.steve.bigdata.flinkmonitor.function;

import com.steve.bigdata.flinkmonitor.model.Event;
import com.steve.bigdata.flinkmonitor.model.WindowResult;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.java.tuple.Tuple2;

/**
 * @Author: STEVE
 * @Description: 聚合函数，用于计算窗口中事件的数值总和和计数
 * @since: 2025-06-26
 */
public class EventAggregator implements AggregateFunction<Event, Tuple2<Double, Long>, WindowResult> {


    /**
     * 创建初始累加器，在窗口初始化时调用，为每个新窗口创建一个干净的累加器
     *
     * @return 初始化的累加器（sum=0.0, count=0L）
     */
    @Override
    public Tuple2<Double, Long> createAccumulator() {
        // Tuple2.of(总和初始值，计数初始值)
        return Tuple2.of(0.0, 0L);
    }

    /**
     * 将事件值累加到中间结果，窗口每收到一个新事件时调用
     *
     * @param event       输入事件对象
     * @param accumulator 当前的累加器状态
     * @return 更新后的累加器
     */
    @Override
    public Tuple2<Double, Long> add(Event event, Tuple2<Double, Long> accumulator) {
        // f0: 累加事件值到总和
        // f1: 增加事件计数
        return Tuple2.of(
                accumulator.f0 + event.getValue(),  // 更新总和
                accumulator.f1 + 1                  // 增加计数
        );
    }

    /**
     * 从累加器获取最终聚合结果，当窗口触发计算时调用，将累加器转换为最终输出格式
     *
     * @param accumulator 最终的累加器状态
     * @return 封装好的窗口结果对象
     */
    @Override
    public WindowResult getResult(Tuple2<Double, Long> accumulator) {
        return new WindowResult(
                null,   // key会在WindowFunction中设置
                0,      // windowStart会在WindowFunction中设置
                0,      // windowEnd会在WindowFunction中设置
                accumulator.f0, // 计算得到的总和
                accumulator.f1  // 计算得到的事件计数
        );
    }

    /**
     * 合并两个累加器（用于会话窗口或故障恢复）
     * 当需要合并多个累加器时调用（如会话窗口合并或从检查点恢复时）
     *
     * @param a 第一个累加器
     * @param b 第二个累加器
     * @return 合并后的累加器
     */
    @Override
    public Tuple2<Double, Long> merge(Tuple2<Double, Long> a, Tuple2<Double, Long> b) {
        return Tuple2.of(
                a.f0 + b.f0,    // 合并总和
                a.f1 + b.f1     // 合并计数
        );
    }

}
